package demo.kafka.streams.web;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/api")
public class WordCountController {

    public static final String TOPIC = "test.stan";

    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public void setKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @PostMapping("/send")
    public String sendMessage(@RequestBody String message) {
        kafkaTemplate.send(TOPIC, message);
        return "Message sent: " + message;
    }

    @GetMapping("/test")
    public String test() throws ExecutionException, InterruptedException {
        // 发送一些测试数据
        String[] testMessages = {
                "hello world kafka streams",
                "spring cloud stream is powerful",
                "kafka streams processing"
        };

        for (String message : testMessages) {
            kafkaTemplate.send(TOPIC, message);
        }

        return "Test messages sent! Check the Kafka topics for results.";
    }
}